package com.flink.flinkdemo.demo;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.function.ToDoubleFunction;

/**
 * @author zhangy on 2020/12/27
 * @description: add you description
 *
 * flink 算子
 *
 * map 即是转换 String ->String ,
 * flatMap String -> 对象，Tuple 或者对象
 *
 * 平均值：计算逻辑，
 *
 */
public class CountDealTime {

    public static void main(String[] args) throws Exception {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> dataSource = env.readTextFile("/Users/zhangyong/Idea-git/flink-d/flink-demo/src/doc/export_1608692734");
        FlatMapOperator<String, Tuple3<String, Integer,Double>> flatMap = dataSource.flatMap(new FlatMapFunction<String, Tuple3<String, Integer,Double>>() {
            @Override
            public void flatMap(String value, Collector<Tuple3<String, Integer,Double>> out) throws Exception {
                String[] split = value.split(",");
                for (String s : split) {
                    if (s.contains("shDealTime")) {
                        String[] words = s.split(":");
                        out.collect(new Tuple3<>(words[0], 1,Double.valueOf(words[1].replace("}", ""))));
                    } else if (s.contains("riskDealTime")) {
                        String[] words = s.split(":");
                        out.collect(new Tuple3<>(words[0], 1,Double.valueOf(words[1].replace("}\"}", ""))));
                    }
                }
            }
        });

        /**
         * 先分组
         */
        ReduceOperator<Tuple3<String, Integer, Double>> reduce = flatMap.groupBy(0).reduce(new ReduceFunction<Tuple3<String, Integer, Double>>() {

            @Override
            public Tuple3<String, Integer, Double> reduce(Tuple3<String, Integer, Double> value1, Tuple3<String, Integer, Double> value2) throws Exception {
                return new Tuple3<>(value1.f0, value1.f1 + value2.f1, value2.f2 + value1.f2);
            }
        });
        FlatMapOperator<Tuple3<String, Integer, Double>, Tuple2<String, Double>> tuple3Tuple2FlatMapOperator = reduce.flatMap(new FlatMapFunction<Tuple3<String, Integer, Double>, Tuple2<String, Double>>() {
            @Override
            public void flatMap(Tuple3<String, Integer, Double> value, Collector<Tuple2<String, Double>> out) throws Exception {
                out.collect(new Tuple2<>(value.f0, Double.valueOf(value.f2 / value.f1)));
            }
        });

        flatMap.print("flatMap");
        reduce.print("reduce");
        tuple3Tuple2FlatMapOperator.print("avg");
    }
}